package com.hkbigdata.userprofile.app

import com.hkbigdata.userprofile.common.bean.TagInfo
import com.hkbigdata.userprofile.common.constcode.ConstCode
import com.hkbigdata.userprofile.common.dao.TagInfoDAO
import com.hkbigdata.userprofile.common.util.MyClickHouseUtil
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

import scala.collection.mutable.ListBuffer

/**
 * @author liuanbo
 * @creat 2024-05-10-15:47
 * @see 2194550857@qq.com
 *
 */
object TaskCkToBitmap {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME", "hkbigdata")
    val conf = new SparkConf().setAppName("task-sql")
      .setMaster("local[*]")
    val sparkSession = SparkSession.builder().config(conf).enableHiveSupport().getOrCreate()


    val date = args(0)
    val infoes = TagInfoDAO.getTaskTagMergeOntask()

    //根据不同的标签类型来创建listbuffer来接收数据
    val infoesListBufferString = new ListBuffer[TagInfo]
    val infoesListBufferdate = new ListBuffer[TagInfo]
    val infoesListBufferDouble = new ListBuffer[TagInfo]
    val infoesListBufferLong = new ListBuffer[TagInfo]

    //匹配infoes 转换成不同的类型储存
    for (elem <- infoes) {
      elem.tagValueType match {
        case ConstCode.TAG_VALUE_TYPE_LONG => infoesListBufferLong.append(elem)
        case ConstCode.TAG_VALUE_TYPE_DATE => infoesListBufferdate.append(elem)
        case ConstCode.TAG_VALUE_TYPE_DECIMAL => infoesListBufferDouble.append(elem)
        case ConstCode.TAG_VALUE_TYPE_STRING => infoesListBufferString.append(elem)
      }
    }

    insertCkToBitmap("user_tag_value_string", infoesListBufferString, date)
    insertCkToBitmap("user_tag_value_decimal", infoesListBufferDouble, date)


    def insertCkToBitmap(tablename: String, infoes: ListBuffer[TagInfo], date: String) {
      /**
       * insert into default.user_tag_value_string
       * select
       * tag.tag_code_value.1 as tag_code,
       * tag.tag_code_value.2 as tag_value,
       * groupBitmapState(uid) as us
       * from (select
       * uid,
       * arrayJoin([('gender',gender),('agegroup',agegroup),('favor',favor)]) as tag_code_value
       * from user_tag_merge) as tag
       * group by
       * tag.tag_code_value.1,
       * tag.tag_code_value.2;
       */
      //删除每天的分区数据
      val deletesql = s"alter table $tablename delete where dt='${date.replace("-", "")}'"
      println(deletesql)
      MyClickHouseUtil.executeSql(deletesql)

      val tupleColum = infoes.map(data => s"('${data.tagCode.toLowerCase()}',${data.tagCode.toLowerCase()})").mkString(",")
      if (infoes.size > 0){
        val insertsql =
          s"""
             |insert into $tablename
             |select
             |tag.tag_code_value.1 as tag_code,
             |tag.tag_code_value.2 as tag_value,
             |groupBitmapState(cast (uid as UInt64 )) as us,
             |'$date' as dt
             |from (select
             |	uid,
             |	arrayJoin([$tupleColum]) as tag_code_value
             |from user_tag_merge_${date.replace("-", "")}) as tag
             |where tag.tag_code_value.2 <> ''
             |group by
             |tag.tag_code_value.1,
             |tag.tag_code_value.2;
             |""".stripMargin
        println(insertsql)

        MyClickHouseUtil.executeSql(insertsql)
      }

    }


  }
}
